/**
 * Copyright (C) 2008 by The Regents of the University of California
 * Redistribution of this file is permitted under the terms of the GNU
 * Public License (GPL).
 *
 * @author Junghoo "John" Cho <cho AT cs.ucla.edu>
 * @date 3/24/2008
 */

#include <cstdio>
#include <map>
#include <iostream>
#include <fstream>
#include "Bruinbase.h"
#include "SqlEngine.h"
#include "BTreeIndex.h"

using namespace std;

// external functions and variables for load file and sql command parsing 
extern FILE* sqlin;
int sqlparse(void);


RC SqlEngine::run(FILE* commandline)
{
  fprintf(stdout, "Bruinbase> ");

  // set the command line input and start parsing user input
  sqlin = commandline;
  sqlparse();  // sqlparse() is defined in SqlParser.tab.c generated from
               // SqlParser.y by bison (bison is GNU equivalent of yacc)

  return 0;
}

RC SqlEngine::select(int attr, const string& table, const vector<SelCond>& cond)
{
  RecordFile rf;   // RecordFile containing the table
  RecordId   rid;  // record cursor for table scanning
  BTreeIndex btidx; // index on the table

  RC     rc;
  int    key;     
  string value;
  int    count = 0;
  int    diff;
  bool   index;

  // open the table file
  if ((rc = rf.open(table + ".tbl", 'r')) < 0) {
    fprintf(stderr, "Error: table %s does not exist\n", table.c_str());
    return rc;
  }

  // Open the index file, if there is one
  if ((rc = btidx.open(table + ".idx", 'r')))
    index = false;
  else
    index = true;

  // If we're using the index 
  if (index) {
    IndexCursor cursor;
    PageId pid;
    int keyLowerBound, keyUpperBound;
    bool keyLowerBounded = false, keyUpperBounded = false;
    bool valueBounded = false;

    // Get condition bounds on keys and whether there is a condition on a value
    for (size_t i = 0; i < cond.size(); i++) {
      // If the condition is on the key
      if (cond[i].attr == 1) {
        int val = atoi(cond[i].value);
        switch (cond[i].comp) {
          case SelCond::EQ:
            // Check if equality bound is already set.
            // If so, there can be no matches.
            if (keyLowerBounded && keyUpperBounded &&
                keyLowerBound == keyUpperBound)
              goto done;

            // Set equality bound
            keyLowerBounded = keyUpperBounded = true;
            keyLowerBound = keyUpperBound = val;
            break;

          case SelCond::LT:
            // If upper bound not set or new upper bound is less than the
            // existing upper bound, set it.
            if (!keyUpperBounded || 
                (keyUpperBounded && (val - 1) < keyUpperBound)) {
              keyUpperBounded = true;
              keyUpperBound = val - 1;
            }
            break;

          case SelCond::GT:
            // If lower bound not set or new lower bound is less than the
            // existing lower bound, set it.
            if (!keyLowerBounded ||
                (keyLowerBounded && (val + 1) > keyLowerBound)) {
              keyLowerBounded = true;
              keyLowerBound = val + 1;
            }
            break;

          case SelCond::LE:
            // If upper bound not set or new upper bound is less than the
            // existing upper bound, set it.
            if (!keyUpperBounded || 
                (keyUpperBounded && val < keyUpperBound)) {
              keyUpperBounded = true;
              keyUpperBound = val;
            }
            break;

          case SelCond::GE:
            // If lower bound not set or new lower bound is less than the
            // existing lower bound, set it.
            if (!keyLowerBounded ||
                (keyLowerBounded && val > keyLowerBound)) {
              keyLowerBounded = true;
              keyLowerBound = val;
            }
            break;

          // SelCond::NE falls under here, since it does not bound anything
          default:
            break;
        }
      }
      // If there is a condition on a value
      else if (cond[i].attr == 2)
        valueBounded = true;
    }

    // Check if resulting bounds are valid
    if (keyLowerBounded && keyUpperBounded && keyLowerBound > keyUpperBound)
      goto done;

    // Set the cursor to the first key to examine
    // If no lower bound set, use the first key, which is at pid=1, eid=0
    if (!keyLowerBounded) {
      cursor.pid = 1;
      cursor.eid = 0;
    }
    // Otherwise try to locate the first key bounded by keyLowerBound
    else {
      if (btidx.locate(keyLowerBound, cursor))
        goto done;
    }

    // Load the leaf node
    pid = cursor.pid;

    // Scan until the end
    while (cursor.pid != -1) {
      // Read the current record
      if ((rc = btidx.readForward(cursor, key, rid)))
        goto exit_select;

      // Enforce upper bound if set
      if (keyUpperBounded && key > keyUpperBound)
        goto done;

      // Get entire tuple if necessary
      if (valueBounded || attr == 2 || attr == 3) {
        // Read tuple
        if ((rc = rf.read(rid, key, value)))
          goto exit_select;
 
        // Check condition on value if any
        if (valueBounded) {
          bool conditionPassed = true;
          for (size_t i = 0; i < cond.size(); i++) {
            if (cond[i].attr == 2) {
              string condValue(cond[i].value);
              switch (cond[i].comp) {
                case SelCond::EQ:
                  if (value != condValue)
                    conditionPassed = false;
                  break;
                
                case SelCond::NE:
                  if (value == condValue)
                    conditionPassed = false;
                  break;

                case SelCond::LT:
                  if (value >= condValue)
                    conditionPassed = false;
                  break;

                case SelCond::GT:
                  if (value <= condValue)
                    conditionPassed = false;
                  break;
  
                case SelCond::LE:
                  if (value > condValue)
                    conditionPassed = false;
                  break;
   
                case SelCond::GE:
                  if (value < condValue)
                    conditionPassed = false;
                  break;
              }
              // If any condition has not passed, no need to check others
              if (!conditionPassed)
                break;
            }
        
            // Don't return the tuple as a result if any condition failed
            if (!conditionPassed)
              continue;
          }
        }
      }
      // Output result
      switch (attr) {
        case 1:  // SELECT key
          fprintf(stdout, "%d\n", key);
          break;
        case 2:  // SELECT value
          fprintf(stdout, "%s\n", value.c_str());
          break;
        case 3:  // SELECT *
          fprintf(stdout, "%d '%s'\n", key, value.c_str());
          break;
      }
      count++;
    }
    goto done;
  }

  // scan the table file from the beginning
  rid.pid = rid.sid = 0;
  count = 0;
  while (rid < rf.endRid()) {
    // read the tuple
    if ((rc = rf.read(rid, key, value)) < 0) {
      fprintf(stderr, "Error: while reading a tuple from table %s\n", table.c_str());
      goto exit_select;
    }

    // check the conditions on the tuple
    for (unsigned i = 0; i < cond.size(); i++) {
      // compute the difference between the tuple value and the condition value
      switch (cond[i].attr) {
      case 1:
	diff = key - atoi(cond[i].value);
	break;
      case 2:
	diff = strcmp(value.c_str(), cond[i].value);
	break;
      }

      // skip the tuple if any condition is not met
      switch (cond[i].comp) {
      case SelCond::EQ:
	if (diff != 0) goto next_tuple;
	break;
      case SelCond::NE:
	if (diff == 0) goto next_tuple;
	break;
      case SelCond::GT:
	if (diff <= 0) goto next_tuple;
	break;
      case SelCond::LT:
	if (diff >= 0) goto next_tuple;
	break;
      case SelCond::GE:
	if (diff < 0) goto next_tuple;
	break;
      case SelCond::LE:
	if (diff > 0) goto next_tuple;
	break;
      }
    }

    // the condition is met for the tuple. 
    // increase matching tuple counter
    count++;

    // print the tuple 
    switch (attr) {
    case 1:  // SELECT key
      fprintf(stdout, "%d\n", key);
      break;
    case 2:  // SELECT value
      fprintf(stdout, "%s\n", value.c_str());
      break;
    case 3:  // SELECT *
      fprintf(stdout, "%d '%s'\n", key, value.c_str());
      break;
    }

    // move to the next tuple
    next_tuple:
    ++rid;
  }

done:
  // print matching tuple count if "select count(*)"
  if (attr == 4) {
    fprintf(stdout, "%d\n", count);
  }
  rc = 0;

  // close the table/index file and return
  exit_select:
  if (index)
    btidx.close();
  rf.close();
  return rc;
}

RC SqlEngine::load(const string& table, const string& loadfile, bool index)
{
  RC rc = 0;

  // Open file for reading as ifstream
  ifstream ifs(loadfile.c_str());
  if (ifs.fail())
    return RC_FILE_OPEN_FAILED;

  // Open RecordFile table.tbl
  RecordFile rf;
  if ((rc = rf.open(table + ".tbl", 'w')))
    return rc;

  // Create index if specified
  BTreeIndex btidx;
  if (index && (rc = btidx.open(table + ".idx", 'w')))
      return rc;

  RecordId rid = rf.endRid();
  map<int, RecordId> tuples;

  // Read tuples line by line
  string tuple;
  while (ifs.good()) {
    getline(ifs, tuple);
    if (tuple.empty())
      continue;

    // Parse load line
    int key;
    string value;
    if ((rc = parseLoadLine(tuple, key, value)))
      return rc;

    // Add line to RecordFile
    if ((rc = rf.append(key, value, rid)))
      return rc;

    // Add (key, rid) pair into the index if specified
    if (index)
      tuples.insert(pair<int, RecordId>(key, rid));

    rid++;
  }

  if (index) {
    for (map<int, RecordId>::iterator it = tuples.begin(); it != tuples.end(); it++) {
      btidx.insert(it->first, it->second);
    }
  }

  // Close index file
  if (index && (rc = btidx.close()))
    return rc;

  return 0;
}

RC SqlEngine::parseLoadLine(const string& line, int& key, string& value)
{
    const char *s;
    char        c;
    string::size_type loc;
    
    // ignore beginning white spaces
    c = *(s = line.c_str());
    while (c == ' ' || c == '\t') { c = *++s; }

    // get the integer key value
    key = atoi(s);

    // look for comma
    s = strchr(s, ',');
    if (s == NULL) { return RC_INVALID_FILE_FORMAT; }

    // ignore white spaces
    do { c = *++s; } while (c == ' ' || c == '\t');
    
    // if there is nothing left, set the value to empty string
    if (c == 0) { 
        value.erase();
        return 0;
    }

    // is the value field delimited by ' or "?
    if (c == '\'' || c == '"') {
        s++;
    } else {
        c = '\n';
    }

    // get the value string
    value.assign(s);
    loc = value.find(c, 0);
    if (loc != string::npos) { value.erase(loc); }

    return 0;
}
